package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/*
 * 此类表示虚拟主机
 * 每个虚拟主机都有自己的 交换机、队列、绑定、消息
 * 相当于进一步封装,类似数据库中的 database
 * 对业务逻辑进行整合,需要对异常进行处理
 */
@Data
@Slf4j
public class VirtualHost {
    private String virtualHostName;
    //硬盘数据管理
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    //内存数据管理
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    //实现路由规则
    private Router router = new Router();

    private final Object exchangeLocker = new Object();

    private final Object queueLocker = new Object();

    private ConsumerManager consumerManager = new ConsumerManager(this);

    public VirtualHost(String name){
        //需要对数据进行初始化(数据库建库建表操作)
        this.virtualHostName = name;
        diskDataCenter.init();
        //将硬盘上的数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MqException | IOException | ClassNotFoundException e) {
            log.error("[VirtualHost] 恢复内存数据失败");
            e.printStackTrace();
        }
    }


    //创建交换
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                   boolean autoDelete, Map<String, Object> arguments) {
        //为了进行虚拟主机的隔离,此处采用在交换机的前缀加上虚拟主机名的方式
        exchangeName = virtualHostName + "_" + exchangeName;
        try{
            synchronized (exchangeLocker){
                //1.判断交换机是否存在
                Exchange exsitsExchange = memoryDataCenter.getExchange(exchangeName);
                if(exsitsExchange != null){
                    throw new MqException("[VirtualHost] 交换机已经存在 exchangeName:" + exchangeName);
                }
                //2.创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.将交换机写入硬盘
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                //4.将交换机写入内存
                memoryDataCenter.insertExchange(exchange);
                log.info("[VirtualHost] 交换机创建成功 exchangeName:" + exchangeName);
                return true;
            }
        } catch (Exception e){
            log.error("[VirtualHost] 交换机创建失败 exchangeName:" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + "_" + exchangeName;
        try{
            synchronized (exchangeLocker){
                //1.判断交换机是否存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange == null){
                    throw new MqException("[VirtualHost] 交换机不存在 exchangeName:" + exchangeName);
                }
                //2.将交换机从硬盘上删除
                //此处不要取反,isDurable 表示 是否存在硬盘上
                //如果对象存储在硬盘上才要删除,没有存在硬盘上就不用
                if(existsExchange.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //3.将交换机从内存中删除
                memoryDataCenter.insertExchange(existsExchange);
                log.info("[VirtualHost] 交换机删除成功 exchangeName:" + exchangeName);
            }
            return true;
        } catch (Exception e){
            log.error("[VirtualHost] 交换机删除失败 exchangeName:" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName, boolean exclusive, boolean durable, boolean autoDelete,
                                Map<String, Object> arguments){
        queueName = virtualHostName + "_" + queueName;
        try{
            synchronized (queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue != null){
                    throw new MqException("[VirtualHost] 队列已经存在 queueName:" + queueName);
                }
                //创建队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //写入硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                //写入内存
                memoryDataCenter.insertQueue(queue);
                log.info("[VirtualHost] 队列创建成功 queueName:" + queueName);
            }
            return true;
        } catch (Exception e){
            log.error("[VirtualHost] 队列创建失败 queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + "_" + queueName;
        try{
            synchronized (queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue == null){
                    throw new MqException("[VirtualHost] 队列不存在 queueName:" + queueName);
                }
                if(existsQueue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                log.error("[VirtualHost] 队列删除成功 queueName:" + queueName);
                return true;
            }
        }catch (Exception e){
            log.error("[VirtualHost] 队列删除失败 queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //添加绑定关系
    public boolean queueBind(String exchangeName, String queueName, String bindingKey){
        exchangeName = virtualHostName + "_" + exchangeName;
        queueName = virtualHostName + "_" + queueName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.验证转发规则
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey非法 bindingKey:" + bindingKey);
                    }
                    //2.判断绑定关系是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existsBinding != null){
                        throw new MqException("[VirtualHost] 绑定已经存在 exchangeName:" + exchangeName + ", queueName:" + queueName);
                    }
                    //3.判断交换机和队列是否都存在
                    Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                    MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                    if(existsExchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在 exchangeName:" + exchangeName);
                    }
                    if(existsQueue == null){
                        throw new MqException("[VirtualHost] 队列不存在 queueName:" + queueName);
                    }
                    //4.创建绑定
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //5.将绑定写入硬盘
                    diskDataCenter.insertBinding(binding);
                    //6.将绑定写入内存
                    memoryDataCenter.insertBinding(binding);
                    log.info("[VirtualHost] 绑定添加成功 exchangeName:" + exchangeName + ", queueName:" + queueName);
                }
            }
            return true;
        } catch (Exception e){
            log.error("[VirtualHost] 绑定创建失败 exchangeName:" + exchangeName + ", queueName:" + queueName);
            e.printStackTrace();
        }
        return false;
    }

    //解除绑定
    public boolean queueUnbind(String exchangeName, String queueName){
        exchangeName = virtualHostName + "_" + exchangeName;
        queueName = virtualHostName + "_" + queueName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //检查是否有这个绑定
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existsBinding == null){
                        throw new MqException("[VirtualHost] 绑定不存在 exchangeName:" + exchangeName + ", queueName:" + queueName);
                    }
                    //从硬盘中删除绑定关系
                    diskDataCenter.deleteBinding(existsBinding);
                    //从内存中删除绑定关系
                    memoryDataCenter.deleteBinding(existsBinding);
                    log.info("[VirtualHost] 绑定解除成功 exchangeName:" + exchangeName + ", queueName:" + queueName);
                }
            }
            return true;
        } catch (Exception e){
            log.error("[VirtualHost] 绑定解除失败 exchangeName:" + exchangeName + ", queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //向指定交换机发送消息,并根据交换机的转发规则,向队列发送消息
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        exchangeName = virtualHostName + "_" + exchangeName;
        try{
            //1.判断交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                throw new MqException("[VirtualHost] 交换机不存在 exchangeName:" + exchangeName);
            }
            //2.判断routingKey是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey非法 routingKey:" + routingKey);
            }
            //3.检查交换机类型
            if(exchange.getType() == ExchangeType.DIRECT){
                //直接交换机, routingKey为queueName
                //4.查找指定队列
                String queueName = virtualHostName + "_" + routingKey;
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 队列不存在 queueName:" + queueName);
                }
                //5.构造message对象
                Message message = Message.createMessageWithId(basicProperties, routingKey, body);
                //6.发送消息到指定队列
                sendMessage(queue, message);
            }else{
                //4.获取到所有绑定关系
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String, Binding> entry : bindings.entrySet()){
                    Binding binding = entry.getValue();
                    //5.查找指定队列
                    String queueName = virtualHostName + "_" + binding.getQueueName();
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        //如果当前不存在,还需要检查其他的队列是否存在
                        continue;
                    }
                    //6.检查路由规则:
                    //      fanout, 则该交换机绑定的所有队列都要转发
                    //      topic, 则需要判断routingKey和bindingKey是否能对上
                    if(!router.route(exchange.getType(), routingKey, binding.getBindingKey())){
                        continue;
                    }
                    //7.构造message对象
                    Message message = Message.createMessageWithId(basicProperties, routingKey, body);
                    //8.发送消息到指定队列
                    sendMessage(queue, message);
                }
                log.info("[VirtualHost] 发送消息成功");
            }
            return true;
        } catch (Exception e){
            log.error("[VirtualHost] 发送消息失败 exchangeName: " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定队列
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //1.写入硬盘
        if(message.isDurable()){
            diskDataCenter.insertMessage(queue, message);
        }
        //2.写入内存
        //将消息写入对应队列
        memoryDataCenter.sendMessage(queue.getName(), message);
        //3.通知消费者消费消息
        consumerManager.notifyConsumer(queue.getName());
    }

    //订阅消息
    //添加一个队列订阅者,当队列收到消息,就把消息推送给订阅者
    //consumerTag: 消息者的唯一标识
    //autoAck: 是否自动应答
    //consumer: 回调函数,并写成了函数式接口,描述了消费者的行为
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        queueName = virtualHostName + "_" + queueName;
        try{
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            log.info("[VirtualHost] 订阅消息成功 queueName:" + consumerTag + ", queueName:" + queueName);
            return true;
        }catch (Exception e){
            log.error("[VirtualHost] 订阅消息失败 consumerTag:" + consumerTag + ", queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //手动应答
    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + "_" + queueName;
        try{
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("[VirtualHost] 要确认的队列不存在 queueName:" + queueName);
            }
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在 messageId:" + messageId);
            }
            if(message.isDurable()){
                diskDataCenter.deleteMessage(queue, message);
            }
            memoryDataCenter.deleteMessageWaitAck(queueName, messageId);
            memoryDataCenter.removeMessage(messageId);
            log.info("[VirtualHost] 消息确认收到 queueName:" + queueName);
            return true;
        }catch (Exception e){
            log.info("[VirtualHost] 消息确认失败 queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

}
